package com.qyer.search.storm.bolt;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import backtype.storm.utils.TupleUtils;
import com.qyer.search.storm.logging.ErrorLogger;
import com.qyer.search.storm.logging.InfoLogger;
import com.qyer.search.storm.logging.TimeCostLogger;
import com.qyer.search.storm.logging.WarnLogger;
import com.qyer.search.storm.util.Channel;
import com.qyer.search.storm.util.CityService;
import com.qyer.search.storm.util.CountryService;
import com.qyer.search.storm.util.ConfigUtils;
import com.qyer.search.storm.util.Country;
import com.qyer.search.storm.util.City;
import com.qyer.search.storm.util.EntityInfo;
import com.qyer.search.storm.util.EntityTags;
import com.qyer.search.storm.util.SetOutPlace;
import com.qyer.search.storm.util.Tag;
import org.apache.kafka.common.utils.CollectionUtils;
import org.apache.storm.shade.org.apache.commons.exec.util.MapUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.qyer.search.storm.util.*;

/**
 * Created by tangzheng on 16/1/8.
 */
public class FetchEntityBolt extends JDBCBolt {
  private static final long serialVersionUID = -1L;
  //private static final Logger LOG = LoggerFactory.getLogger(FetchEntityBolt.class);
  private String _streamId;
  protected CommonCountryService _commonCountryService;
  protected CommonCityService _commonCityService;


  public FetchEntityBolt(String boltPrefix, int inCount, int inCount2, String streamId,
                         Map<String, String> sqlConfig){
    super(boltPrefix, inCount, inCount2, sqlConfig);
    this._streamId = streamId;
  }

  @Override public void execute(Tuple tuple) {
    //currentTime(_prefix + ",execute in,");
    if (TupleUtils.isTick(tuple)) {
      //LOG.info("Tick arrive.");
      tickInfoLogger.info("Tick arrive.");
      checkDBConnection();
    }
    else {
      checkDBConnection();

      long now = System.currentTimeMillis();

      String index = tuple.getString(0);
      String type = tuple.getString(1);
      List<Integer> payload = (List<Integer>) tuple.getValue(2);
      int msgCount = tuple.getInteger(3);
      int stage = tuple.getInteger(4);
      PreparedStatement pstmt_in_use;
      int inCount_in_use = 0;
      //******************************************************DB
      /*
      LOG.info("--------------------fetchInfoIn:{}------------------------", msgCount);
      LOG.info("payload:{}", payload);
      LOG.info("index  :{}", index);
      LOG.info("type   :{}", type);
      LOG.info("msgId  :{}", msgCount);
      LOG.info("stage  :{}", stage);
      LOG.info("--------------------fetchInfoIn:{}------------------------", msgCount);
      */

      ///////////////////////////////////////////////////////////
      dataLog.info("{},msgId:{},index:{},type:{},stage:{},payload:{}", ConfigUtils.IN, msgCount, index, type, stage, payload);
      ///////////////////////////////////////////////////////////

      pstmt_in_use = (ConfigUtils.INIT_STAGE == stage) ? _pstmt : _pstmt2;
      inCount_in_use = (ConfigUtils.INIT_STAGE == stage) ? _inCount : _inCount2;

      /*if(ConfigUtils.INIT_STAGE == stage) {
        pstmt_in_use = _pstmt;
        inCount_in_use = _inCount;
      }
      else {
        pstmt_in_use = _pstmt2;
        inCount_in_use = _inCount2;
      }

      pstmt_in_use = _pstmt;
      inCount_in_use = _inCount;*/

      try {
        ConfigUtils.constructSQL(pstmt_in_use, payload, inCount_in_use);
      } catch (Exception e) {
        //LOG.error("msgId:{},constructSQL failed,{}", msgCount, e);
        dbInfoLogger.info("msgId:{},constructSQL failed,{}", msgCount, e);
      }

      /*
      select
      id, title, product_type, product_child_type, cityids, countryids,
      channel, supplierid, travel_starttime, travel_endtime, abroad
      from lastminute_main where id in(?,?,?,?,?);
      */
      int lid;
      int product_type;
      int product_child_type;

      String title;
      String cityids;
      String countryids;
      String channel;

      int supplierid;
      long travel_starttime;
      long travel_endtime;
      String abroad;

      Map<Integer, EntityInfo> infoMap = new HashMap<>();

      try {
        ResultSet rs = pstmt_in_use.executeQuery();
        while (rs.next()) {
          lid = rs.getInt(1);
          title = rs.getString(2).trim();
          product_type = rs.getInt(3);
          product_child_type = rs.getInt(4);
          cityids = rs.getString(5).trim();
          countryids = rs.getString(6).trim();
          channel = rs.getString(7).trim();

          supplierid = rs.getInt(8);
          travel_starttime = rs.getLong(9);
          travel_endtime = rs.getLong(10);
          abroad = rs.getString(11).trim();
          processLogger.info("msgId:{},product id:{},title:{},cityids:{},countryids:{}," +
                               "product_type:{},product_child_type:{},supplierid:{},travel_starttime:{}" +
                               ",travel_endtime:{},abroad:{}", msgCount, lid, title, cityids, countryids,
                             product_type, product_child_type, supplierid, travel_starttime, travel_endtime,
                             abroad);

          if (cityids.startsWith(",") || cityids.startsWith("，") || countryids.startsWith(",") || countryids.startsWith("，") || channel.startsWith(",") || channel
            .startsWith("，") || abroad.startsWith(",") || abroad.startsWith("，")) {
            //LOG.warn("invalid city({})/country({})/channel({}) format.", cityids, countryids,
                     //channel);
            errorLogger.info("msgId:{},invalid city({})/country({})/channel({}/abroad{}) format.", msgCount, cityids, countryids,
                             channel, abroad);
            //continue;
          }

          cityids = cityids.replaceAll("[ ]*[，,\\,][ ]*", ",");
          countryids = countryids.replaceAll("[ ]*[，,\\,][ ]*", ",");
          channel = channel.replaceAll("[ ]*[，,\\,][ ]*", ",");
          abroad = abroad.replaceAll("[ ]*[，,\\,][ ]*", ",");


          String[] cityArr = cityids.split(",");
          String[] countryArr = countryids.split(",");
          String[] channelArr = channel.split(",");
          String[] abroadArr = abroad.split(",");


          List<Country> countryList = new ArrayList<>();
          List<City> cityList = new ArrayList<>();
          List<Channel> channelList = new ArrayList<>();
          List<SetOutPlace> setOutPlaceList = new ArrayList<>();


          for (int i = 0; i < cityArr.length; i++) {
            String cityStr = cityArr[i].trim();
            if(cityStr.matches("[0-9]+")) {
              int id = Integer.valueOf(cityArr[i]);
              //String name = CityService.getInstance(_sqlConfig, _boltPrefix).getCityName(id);
              String name = _commonCityService.getCityName(id);
              City city = new City(id, name);
              cityList.add(city);
            }
          }
          for (int i = 0; i < countryArr.length; i++) {
            String countryStr = countryArr[i].trim();
            if(countryStr.matches("[0-9]+")) {
              int id = Integer.valueOf(countryStr);
              //String name = CountryService.getInstance(_sqlConfig, _boltPrefix).getCountryName(id);
              String name = _commonCountryService.getCountryName(id);
              Country country = new Country(id, name);
              countryList.add(country);
            }
          }
          for (int i = 0; i < channelArr.length; i++) {
            if(!"".equals(channelArr[i])) {
              channelList.add(new Channel(channelArr[i]));
            }
          }

          for(int i =0; i < abroadArr.length; i++){

            if(ConfigUtils.START_POS_Map.containsKey(abroadArr[i])){
              setOutPlaceList.add(new SetOutPlace(abroadArr[i], ConfigUtils.START_POS_Map.get(abroadArr[i])));
            }
          }


          if(cityList.size() == 0){
            cityList = null;
          }
          if(countryList.size() == 0){
            countryList = null;
          }
          if(channelList.size() == 0){
            channelList = null;
          }
          if(setOutPlaceList.size() == 0){
            setOutPlaceList = null;
          }

          EntityInfo ei = new EntityInfo(lid, title, product_type, product_child_type, supplierid,
                                         travel_starttime, travel_endtime, setOutPlaceList, cityList,
                                         countryList, channelList);

          processLogger.info("msgId:{},product id:{},title:{},cityList:{},countryList:{},product_type:{},product_child_type:{}," +
                               "supplierid:{},travel_starttime:{},travel_endtime:{},setOutPlaceList:{}"
            , msgCount, lid, title, cityList, countryList, product_type, product_child_type,
                             supplierid, travel_starttime, travel_endtime, setOutPlaceList);

          infoMap.put(lid, ei);
        }

        long dbTime = System.currentTimeMillis();
        int dbCost = (int)(dbTime - now);

        //currentTime(_prefix + ",execute db out,");
        if(infoMap == null) {
          infoMap = new HashMap<>();
        }
        String infoStr = ConfigUtils.DEFAULT_GSON.toJson(infoMap);
        //currentTime(_prefix + "execute out,");
        _outputCollector.emit(_streamId, tuple, new Values(index, type, infoStr, msgCount));

        /*
        LOG.info("---------------------fetchInfoOut:{}-----------------------", msgCount);
        LOG.info("index  :{}", index);
        LOG.info("type   :{}", type);
        LOG.info("infoStr  :{}", infoStr);
        LOG.info("msgCount  :{}", msgCount);
        LOG.info("---------------------fetchInfoOut:{}-----------------------", msgCount);
        */

        ///////////////////////////////////////////////////////
        dataLog.info("{},msgId:{},index:{},type:{},stage:{},payload:{}", ConfigUtils.OUT, msgCount, index, type, stage, infoStr);
        ///////////////////////////////////////////////////////

        _outputCollector.ack(tuple);

        ////////////////////////////////////////
        long end = System.currentTimeMillis();
        int cost = (int)(end - now);
        ////////////////////////////////////////

        timeCostLogger.info("{}:{}:{}:{}:{}:{}", msgCount, now, dbTime, end, dbCost, cost);
      } catch (Exception e) {
        //LOG.error("DB get info failed,{}", e.getMessage());
        //LOG.error("msgCount:{},payload:{}", msgCount, payload);

        errorLogger.info("msgId:{},DB get entity info failed,{}", msgCount, e);
        errorLogger.info("msgId:{},payload:{}", msgCount, payload);

        _outputCollector.fail(tuple);
      }

      //******************************************************DB

      //currentTime(_prefix + "execute out,");
    }
  }

  @Override public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    outputFieldsDeclarer.declareStream(_streamId, new Fields("index", "type", "info", "msgId"));
  }

  @Override public void prepare(Map map, TopologyContext topologyContext,
                                OutputCollector outputCollector){

    String directory = "/Data/search/FetchEntityBolt/";
    String filename = directory + topologyContext.getStormId()+"/"
      + topologyContext.getThisComponentId() +"/"+topologyContext.getThisTaskId();
    String loggername = JDBCBolt.class.getSimpleName() + "@" + topologyContext.getThisTaskId();
    dataLog = new InfoLogger(filename + "_data", loggername + "_data");
    infoLogger = new InfoLogger(filename + "_info", loggername + "_info");
    errorLogger = new ErrorLogger(filename +"_error", loggername + "_error");
    timeCostLogger = new TimeCostLogger(filename + "_timecost", loggername + "_timecost");
    successLogger = new WarnLogger(filename + "_success", loggername + "_success");
    failLogger = new WarnLogger(filename + "_fail", loggername + "_fail");
    dbInfoLogger = new InfoLogger(filename + "_db", loggername + "_db");
    tickInfoLogger = new InfoLogger(filename + "_tick", loggername + "_tick");
    processLogger = new InfoLogger(filename + "_process", filename + "_process");

    String storminfo = topologyContext.getStormId()+"_"
      + topologyContext.getThisComponentId() +"_"+topologyContext.getThisTaskId();
    _commonCityService = CommonCityService.getInstance(_sqlConfig, _boltPrefix, storminfo);
    _commonCountryService = CommonCountryService.getInstance(_sqlConfig, _boltPrefix, storminfo);

    _topologyContext = topologyContext;
    _outputCollector = outputCollector;
    _config = (Map<String, Object>)map;
    _taskId = topologyContext.getThisTaskId();
    _componentId = topologyContext.getThisComponentId();
    _prefix = _boltPrefix + ",TaskId:"+ _taskId + ",ComponentId:" + _componentId;
    currentTime(_prefix + "Bolt prepare");

    try {
      _conn = ConfigUtils.getConnection((Map<String, Object>) _sqlConfig, _boltPrefix);
      //LOG.info("{} DB init successfully!", _boltPrefix);
      dbInfoLogger.info("{} DB init successfully!", _boltPrefix);
    }catch (Exception e){
      //LOG.error("{} Connect DB failed.", _boltPrefix);
      dbInfoLogger.info("{} Connect DB failed.", _boltPrefix);
    }

    try {
      _pstmt = ConfigUtils.getPrepareStmt((Map<String, Object>) map, _boltPrefix, this._conn);
      _pstmt2 = ConfigUtils.getPrepareStmt2((Map<String, Object>) map, _boltPrefix, this._conn);
    }catch (Exception e){
      //LOG.error("{} get PreparedStatement failed.", _boltPrefix);
      dbInfoLogger.info("{} get PreparedStatement failed.", _boltPrefix);
    }


  }


  @Override public void cleanup() {
    //currentTime(_prefix + "cleanup,");
    processLogger.info("begin cleanup.");
    try {
      if( _pstmt != null)
        _pstmt.close();
      if(_pstmt2 != null)
        _pstmt2.close();
      if(_conn != null)
        _conn.close();
      dbInfoLogger.info("mysqlconnection closed.");
      processLogger.info("mysqlconnection closed.");
      //LOG.info("mysqlconnection closed");
    }catch (Exception e){
      dbInfoLogger.info(_prefix + e.getMessage());
      //LOG.error(_prefix + e.getMessage());
    }

    if(_commonCountryService != null)
      _commonCountryService.shutdown();
    processLogger.info("CommonCountryService shutdown.");

    if(_commonCityService != null)
      _commonCityService.shutdown();
    processLogger.info("CommonCityService shutdown.");
  }
}
